package com.bw.gmall.realtime.app.dwd;

import com.bw.gmall.realtime.utils.MyKafkaUtil;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;

public class DwdToolCouponPay {
    public static void main(String[] args) {
        //todo 初始化上下文  设置并行度
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);
        StreamTableEnvironment tenv = StreamTableEnvironment.create(env);


        //TODO 3. 从 Kafka 读取业务数据，封装为 Flink SQL 表

        tenv.executeSql("create table `topic_db` (" +
                "`database` string, " +
                "`table` string, " +
                "`type` string, " +
                "`data` map<string,string>, " +
                "`old` map<string,string>, " +
                "`ts` string " +
                ")"+ MyKafkaUtil.getKafkaDDL("topic_db","dwd_tool_coupon_pay"));

        // TODO 4. 读取优惠券领用表数据，筛选优惠券使用（支付）数据
        Table couponUsePay = tenv.sqlQuery("select " +
                "data['id'] id, " +
                "data['coupon_id'] coupon_id, " +
                "data['user_id'] user_id, " +
                "data['order_id'] order_id, " +
                "date_format(data['used_time'],'yyyy-MM-dd') date_id, " +
                "data['used_time'] used_time, " +
                "`old`, " +
                "ts " +
                "from topic_db " +
                "where `table` = 'coupon_use' " +
                "and `type` = 'update' " +
                "and data['used_time'] is not null ");
        tenv.createTemporaryView("coupon_use_pay",couponUsePay);

        // TODO 5. 建立 Kafka-Connector dwd_tool_coupon_pay 表
        tenv.executeSql("create table dwd_tool_coupon_pay (" +
                "id string, " +
                "coupon_id string, " +
                "user_id string, " +
                "order_id string, " +
                "date_id string, " +
                "payment_time string, " +
                "ts string " +
                ")" + MyKafkaUtil.getKafkaSinkDDL("dwd_tool_coupon_pay"));

        //todo 6.将数据写入kafka主题中
        tenv.executeSql("insert into dwd_tool_coupon_pay select " +
                "id, " +
                "coupon_id, " +
                "user_id, " +
                "order_id, " +
                "date_id, " +
                "used_time payment_time, " +
                "ts " +
                "from coupon_use_pay ");


    }
}
